#ifndef __MQ_CONSUME_HPP__
#define __MQ_CONSUME_HPP__

#include "mq_exchange.hpp"
#include "mq_queue.hpp"
#include <iostream>
#include <string>
#include <unordered_map>
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <cassert>
#include <memory>
#include <mutex>
#include <functional>

namespace mq
{
    using ConsumerCallback = std::function<void(const std::string&,const BasicProperties* ,const std::string&)>;
    struct Consumer
    {
        using Ptr = std::shared_ptr<Consumer>;
        std::string tag; // 消费者标识。
        std::string qname; // 消费者订阅的队列名称。
        bool auto_ack; // 自动确认标志。
        ConsumerCallback callback;

        // 无参构造，方便指针构造对象。
        Consumer() { DLOG("无参构造出Consumer"); }
        ~Consumer() { DLOG("Consumer析构"); }
        Consumer(const std::string& ctag,const std::string& queue_name,bool ack_flag,const ConsumerCallback& cb)
            :tag(ctag)
            ,qname(queue_name)
            ,auto_ack(ack_flag)
            ,callback(std::move(cb) )
        {
            DLOG("带参构造出Consumer tag:%s",tag.c_str());
        }
    };

    // 以队列为单元的消费者管理结构。
    class QueueConsumer
    {
        public:
            using Ptr = std::shared_ptr<QueueConsumer>;
            QueueConsumer(const std::string& qname)
                :_qname(qname)
            { DLOG("Consumer Queue Name:%s",qname.c_str()); }
            ~QueueConsumer()
            { DLOG("Delete Consumer Queue Name:%s",_qname.c_str()); }
            // 队列新增消费者。
            Consumer::Ptr create(const std::string& ctag,
                const std::string& queue_name,
                bool ack_flag,
                const ConsumerCallback& cb)
            {
                // 1. 老步骤：加锁
                std::unique_lock<std::mutex> lock(_mutex);

                DLOG("in consumer queue");
                // 2. 寻找是否存在重名，存在则返回空指针。
                for(auto& consumer : _consumers)
                {
                    if (consumer->tag == ctag)
                    {
                        DLOG("存在该标签：%s",ctag.c_str());
                        return Consumer::Ptr();
                    }
                }
                // 3. 没有重名，则构建对象。
                // DLOG("构建消费者队列:ctag:%s \t qname:%s",ctag.c_str(),queue_name.c_str());
                Consumer::Ptr consumer = std::make_shared<Consumer>(ctag,queue_name,ack_flag,std::move(cb));
        
                // DLOG("consumer:%p",consumer.get());
                
                // 4.添加进入数组。
                // DLOG("before insert size:%ld",_consumers.size());
                _consumers.push_back(consumer);
                // DLOG("before insert size:%ld",_consumers.size());

                return consumer;
            }
            // 队列移除消费者。
            void remove(const std::string& ctag)
            {
                // 加锁加锁、还是要加锁。
                std::unique_lock<std::mutex> lock(_mutex);
                for(auto it = _consumers.begin();it != _consumers.end();++it)
                {
                    if((*it)->tag == ctag)
                    {
                        DLOG("找到要删除的元素了：%s",ctag.c_str());
                        _consumers.erase(it);

                        return ;
                    }
                }
                DLOG("没有找到要删除的元素了：%s",ctag.c_str());

                return ;
            }
            // 队列获取消费者:RR轮转获取。
            Consumer::Ptr choose()
            {
                std::unique_lock<std::mutex> lock(_mutex);
                if(_consumers.size() == 0)
                {
                    DLOG("choose数组内无元素。");
                    
                    return Consumer::Ptr();
                }
                int index = _rr_seq;
                ++_rr_seq;
                _rr_seq %= _consumers.size();

                return _consumers[index];
            }
            // 是否为空。
            bool empty()
            {
                std::unique_lock<std::mutex> lock(_mutex);

                return _consumers.size() == 0;
            }
            // 判断指定消费者是否存在。
            bool exists(const std::string& ctag)
            {
                std::unique_lock<std::mutex> lock(_mutex);

                DLOG("开始寻找:%s",ctag.c_str());
                for(auto it = _consumers.begin();it != _consumers.end();++it)
                {
                    if((*it)->tag == ctag)
                    {
                        DLOG("存在该元素:%s",ctag.c_str());
                        return true;
                    }
                }
                DLOG("不存在该元素:%s",ctag.c_str());

                return false;
            }
            // 清理所有消费者。
            void clear()
            {
                std::unique_lock<std::mutex> lock(_mutex);

                DLOG("Consumers已清理");
                _consumers.clear();
                _rr_seq = 0;
            }
        private:
            std::mutex _mutex;
            std::string _qname;
            uint64_t _rr_seq; // 轮转序号
            std::vector<Consumer::Ptr> _consumers;
    };
    class ConsumerManager
    {
    public:
        using Ptr = std::shared_ptr<ConsumerManager>;
        ConsumerManager() { DLOG("consume_manager:%p",this); }
        void initQueueConsumer(const std::string& qname)
        {
            DLOG("initQueueConsumer");
            std::unique_lock<std::mutex> lock(_mutex);
            
            auto it = _qconsumers.find(qname);
            // 存在消费者队列，无需初始化。
            if(it != _qconsumers.end())
            {
                DLOG("initQueueConsumer exists: %s , no init.",qname.c_str());

                return ;
            }
            DLOG("New QueueConsumer");
            QueueConsumer::Ptr qcp = std::make_shared<QueueConsumer>(qname);
            _qconsumers.insert(std::make_pair(qname,qcp));

            DLOG("qcp: %p",qcp.get());

            DLOG("new insert _qconsumers: %p , name: %s",_qconsumers[qname].get(),qname.c_str());

            return ;
        }
        void destroyQueueConsumer(const std::string& qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            
            _qconsumers.erase(qname);
        }
        Consumer::Ptr create(const std::string& ctag,const std::string& queue_name,bool ack_flag,const ConsumerCallback& cb)
        {
            DLOG("in ConsumeManager create");
            DLOG("tag:%s \t queue_name:%s \t ack_flag:%d ",
                ctag.c_str(),
                queue_name.c_str(),
                ack_flag);
            QueueConsumer::Ptr qcp;
            // 锁只保护当前的共享资源，一经取出概不负责。
            // 其他的共享资源由其他锁保护。
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(queue_name);
                if(it == _qconsumers.end())
                {
                    DLOG("在消费者队列中该队列%s不存在",queue_name.c_str());

                    return Consumer::Ptr();
                }
                qcp = it->second;
            }
            // DLOG("在ConsumeManager中的create函数中成功创建一个consumer对象");

            return qcp->create(ctag,queue_name,ack_flag,std::move(cb));
        }
        void remove(const std::string& ctag,const std::string& queue_name) 
        {
            QueueConsumer::Ptr qcp;
            // 锁只保护当前的共享资源，一经取出就不管了。
            // 其他的共享资源由其他锁保护。
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(queue_name);
                if(it == _qconsumers.end())
                {
                    DLOG("该消息不存在%s，不用删除",ctag.c_str());
                    return ;
                }
                qcp = it->second;
            }
            
            return qcp->remove(ctag);
        }
        Consumer::Ptr choose(const std::string& queue_name)
        {
            QueueConsumer::Ptr qcp;
            int i = 0;
            for(auto& consumer : _qconsumers) // *it not it
            {
                DLOG("Queue Consumer[%d]:%s",i,consumer.first.c_str());
                ++i;
            }
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(queue_name);
                if(it == _qconsumers.end())
                {
                    DLOG("队列%s信息未能找到",queue_name.c_str());

                    return Consumer::Ptr();
                }
                qcp = it->second;
            }

            return qcp->choose();
        }
        bool empty(const std::string& queue_name)
        {
            QueueConsumer::Ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(queue_name);
                if(it == _qconsumers.end())
                {
                    DLOG("队列不存在%s",queue_name.c_str());
                    return false;
                }
                qcp = it->second;
            }
            
            return qcp->empty();
        }
        bool exists(const std::string& ctag,const std::string& queue_name)
        {
            QueueConsumer::Ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(queue_name);
                if(it == _qconsumers.end())
                {
                    DLOG("队列不存在%s",queue_name.c_str());

                    return false;
                }
                DLOG("队列存在%s",queue_name.c_str());
                qcp = it->second;
            }

            return qcp->exists(ctag);
        }
        size_t size()
        {
            std::unique_lock<std::mutex> lock(_mutex);

            return _qconsumers.size();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _qconsumers.clear();
        }
    private:
        std::mutex _mutex;
        std::unordered_map<std::string,QueueConsumer::Ptr> _qconsumers;
    };
}

#endif